package com.mjf.gmall.realtime.dws

import java.{lang, util}

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import com.mjf.gmall.realtime.bean.{OrderDetail, OrderDetailWide, OrderInfo}
import com.mjf.gmall.realtime.util.{MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * orderInfo 与 orderDetail join 生成宽表(使用滑动窗口实现)
 *
 * 技术栈：
 *  zookeeper/kafka/redis/
 *
 * -双流join问题：
 *  无法保证应当配对的主表和从表数据都在一个批次中，join有可能丢失数据
 *
 * -解决方案：
 *  1.使用缓存
 *  2.使用滑动窗口+数据去重（适用于短延迟，延迟越大数据膨胀越大）
 *
 *
 * -注意：
 *  join时尽量不要出现shuffle。
 * -解决方案：
 *  在join前的数据保证分区是一对一的关系，利用kafka发送时的分区键，两张表的分区键和分区数保持一致。
 */
object OrderInfoDetailWideApp {
  def main(args: Array[String]): Unit = {
    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("order_wide_app")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topicOrderInfo = "dwd_order_info"
    val groupIdOrderInfo = "dws_order_info_group"

    val topicOrderDetail = "dwd_order_detail"
    val groupIdOrderDetail = "dws_order_detail_group"

    // 从redis读取offset - orderInfo
    val orderInfoKafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topicOrderInfo, groupIdOrderInfo)

    // 从redis读取offset - orderDetail
    val orderDetailKafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topicOrderDetail, groupIdOrderDetail)


    // 从Kafka消费数据 - orderInfo
    var orderInfoRecordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderInfoKafkaOffsetMap == null && orderInfoKafkaOffsetMap.size >0) {
      // 如果redis中已经有了offset信息
      orderInfoRecordInputStream = MyKafkaUtil.getKafkaStream(topicOrderInfo, ssc, orderInfoKafkaOffsetMap, groupIdOrderInfo)
    } else {
      orderInfoRecordInputStream = MyKafkaUtil.getKafkaStream(topicOrderInfo, ssc, groupIdOrderInfo)
    }

    // 从Kafka消费数据 - orderDetail
    var orderDetailRecordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (orderDetailKafkaOffsetMap == null && orderDetailKafkaOffsetMap.size >0) {
      // 如果redis中已经有了offset信息
      orderDetailRecordInputStream = MyKafkaUtil.getKafkaStream(topicOrderDetail, ssc, orderInfoKafkaOffsetMap, groupIdOrderDetail)
    } else {
      orderDetailRecordInputStream = MyKafkaUtil.getKafkaStream(topicOrderDetail, ssc, groupIdOrderDetail)
    }


    // 获得偏移结束点 - orderInfo
    var orderInfoOffsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  // 写在算子外面为启动时执行
    val orderInfoOffsetDstream: DStream[ConsumerRecord[String, String]] = orderInfoRecordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  // 写在算子里面是周期性执行（driver端）
        rdd
    }

    // 获得偏移结束点 - orderDetail
    var orderDetailOffsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  // 写在算子外面为启动时执行
    val orderDetailOffsetDstream: DStream[ConsumerRecord[String, String]] = orderDetailRecordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  // 写在算子里面是周期性执行（driver端）
        rdd
    }


    // 将json字符串 转换为 OrderInfo 对象
    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetDstream.map{
      record =>
        val jsonString: String = record.value()
        val orderInfo: OrderInfo = JSON.parseObject(jsonString, classOf[OrderInfo])

        orderInfo
    }

    // 将json字符串 转换为 OrderDetail 对象
    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetDstream.map{
      record =>
        val jsonString: String = record.value()
        val orderDetail: OrderDetail = JSON.parseObject(jsonString, classOf[OrderDetail])

        orderDetail
    }


    orderInfoDStream.print(1000)
    orderDetailDStream.print(1000)

    // 转换成kv结构
    val orderInfoWithKeyDStream: DStream[(Long, OrderInfo)] = orderInfoDStream.map(orderInfo => (orderInfo.id, orderInfo))
    val orderDetailWithKeyDStream: DStream[(Long, OrderDetail)] = orderDetailDStream.map(orderDetail => (orderDetail.order_id, orderDetail))

    // 无法保证应当配对的主表和从表数据都在一个批次中，join有可能丢失数据
//    val orderJoinedDStream: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoWithKeyDStream.join(orderDetailWithKeyDStream)
//    orderJoinedDStream.print(1000)


    // 使用滑动窗口进行双流join
    // 1.开窗口
    val orderInfoWithKeyWindowDStream: DStream[(Long, OrderInfo)] = orderInfoWithKeyDStream.window(Seconds(10), Seconds(5))
    val orderDetailWithKeyWindowDStream: DStream[(Long, OrderDetail)] = orderDetailWithKeyDStream.window(Seconds(10), Seconds(5))

    // 2.join 有重复数据
    val orderJoinedDStream: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoWithKeyWindowDStream.join(orderDetailWithKeyWindowDStream)

    // 3.去重
    val orderJoinNewDStream: DStream[(Long, (OrderInfo, OrderDetail))] = orderJoinedDStream.mapPartitions {
      orderJoinedIter =>
        val jedis: Jedis = RedisUtil.getJedisClient
        val key = "order_join_keys"
        val orderJoinedList: ListBuffer[(Long, (OrderInfo, OrderDetail))] = new ListBuffer[(Long, (OrderInfo, OrderDetail))]

        for ((orderId, (orderInfo, orderDetail)) <- orderJoinedIter) {
          // redis(type:set key:order_join_keys value:orderDetail.id) ---> orderDetail.id不应该重复
          val ifNew: lang.Long = jedis.sadd(key, orderDetail.id.toString)
          if (ifNew == 1L) {
            orderJoinedList.append((orderId, (orderInfo, orderDetail)))
          }
        }

        jedis.close()
        orderJoinedList.toIterator
    }

    val orderDetailWideDStream: DStream[OrderDetailWide] = orderJoinNewDStream.map {
      case (orderId, (orderInfo, orderDetail)) =>
        new OrderDetailWide(orderInfo, orderDetail)
    }



    // 计算实付分摊需求
    val orderDetailWideWithSplitDStream: DStream[OrderDetailWide] = orderDetailWideDStream.mapPartitions {
      orderDetailWideIter =>
        val orderDetailWideList: List[OrderDetailWide] = orderDetailWideIter.toList

        // 1.先从redis中取两个合计  【实付分摊金额】的合计、【数量*单价】的合计
        val jedis: Jedis = RedisUtil.getJedisClient
        for (orderDetailWide <- orderDetailWideList) {
          // type?hash  key?order_split_amount:[order_id] field?split_amount_sum,origin_amount_sum  value?累计金额
          val key: String = "order_split_amount:[" + orderDetailWide.order_id + "]"
          val orderSumMap: util.Map[String, String] = jedis.hgetAll(key)
          var splitAmountSum: Double = 0D // 实付分摊累计金额
          var originAmountSum: Double = 0D // 原始累计金额
          if (orderSumMap != null && orderSumMap.size() > 0) {
            val splitAmountSumString: String = orderSumMap.get("split_amount_sum")
            splitAmountSum = splitAmountSumString.toDouble

            val originAmountSumString: String = orderSumMap.get("origin_amount_sum")
            originAmountSum = originAmountSumString.toDouble
          }

          // 2.先判断是否最后一笔：(数量*单价) == 原始总金额 - (其他明细【数量*单价】的总计)
          val detailOriginAmount: Double = orderDetailWide.sku_num * orderDetailWide.sku_price // 单笔的原始金额
          val restOriginAmount: Double = orderDetailWide.original_total_amount - originAmountSum // 剩余的原始金额
          if (detailOriginAmount == restOriginAmount) {
            // 3.1最后一笔用减法:实付分摊金额 = 实付总金额 - (其他明细已经计算好的【实付分摊金额】的合计)
            orderDetailWide.final_detail_amount = orderDetailWide.final_total_amount - splitAmountSum
          } else {
            // 3.2不是最后一笔用乘除:实付分摊金额 = (数量*单价) * 实付总金额 / 原始总金额
            orderDetailWide.final_detail_amount = detailOriginAmount * orderDetailWide.final_total_amount / orderDetailWide.original_total_amount
            orderDetailWide.final_detail_amount = Math.round(orderDetailWide.final_detail_amount * 100D) / 100D // 对结果保留两位小数
          }

          // 4.进行合计保存:【实付分摊金额】的合计、【数量*单价】的合计
          splitAmountSum += orderDetailWide.final_detail_amount
          originAmountSum += detailOriginAmount
          orderSumMap.put("split_amount_sum", splitAmountSum.toString)
          orderSumMap.put("origin_amount_sum", originAmountSum.toString)
          jedis.hmset(key, orderSumMap)

        }

        jedis.close()
        orderDetailWideList.toIterator
    }

    orderDetailWideWithSplitDStream.print(1000)

    ssc.start()
    ssc.awaitTermination()


  }
}
